package com.banmajio.service.impl;

import com.banmajio.bean.Kdp01;
import com.banmajio.config.BatchCopyHelper;
import com.banmajio.config.SequenceGenerator;
import com.banmajio.entity.VolunteerInfo;
import com.banmajio.entity.VolunteerInfoDTO;
import com.banmajio.mapper.Kdp01Mapper;
import com.banmajio.mapper.XtVolunteerInfoMapper;
import com.banmajio.mapper.YoungsterInfoMapper;
import com.banmajio.pojo.VolunteerTrajectory;
import com.banmajio.pojo.XtVolunteerInfo;
import com.banmajio.pojo.YoungsterInfo;
import com.banmajio.service.DataCleanningService;
import com.banmajio.util.DataCleanUtils;
import com.banmajio.util.SnowflakeIdGeneratorUtils;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StopWatch;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Service
public class DataCleanningServiceImpl extends ServiceImpl<XtVolunteerInfoMapper, XtVolunteerInfo> implements DataCleanningService {

    private static final Logger log = LoggerFactory.getLogger(DataCleanningServiceImpl.class);
    private static final Pattern TABLE_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9_]+$");
    private final ThreadLocal<Set<String>> processedIdNumbers = ThreadLocal.withInitial(HashSet::new);

    @Value("${data.clean.pageSize:10000}")
    private int pageSize;

    @Value("${data.clean.batchSize:10000}")
    private int batchSize;

    @Value("${data.clean.threadCount:0}")
    private Integer threadCount;

    @Value("${data.clean.maxQueueSize:100}")
    private int maxQueueSize;

    @Autowired
    private Kdp01Mapper kdp01Mapper;

    @Autowired
    private XtVolunteerInfoMapper xtVolunteerInfoMapper;

    @Autowired
    private YoungsterInfoMapper youngsterInfoMapper;

    @Autowired
    private BatchCopyHelper batchCopyHelper;

    @Autowired
    private SequenceGenerator sequenceGenerator;

    private ExecutorService executorService;
    private BlockingQueue<List<Map<String, Object>>> dataQueue;
    private CountDownLatch processLatch;
    private AtomicLong totalProcessedCount = new AtomicLong(0);
    private AtomicLong totalBatchCount = new AtomicLong(0);

    @PostConstruct
    public void init() {
        // 初始化线程数
        this.threadCount = threadCount == 0 ?
                Runtime.getRuntime().availableProcessors() * 2 + 1 : threadCount;

        // 初始化线程池（不立即创建，在execute时创建）
        log.info("数据清洗服务初始化完成，配置: 核心线程数: {}, 批次大小: {}, 页面大小: {}",
                threadCount, batchSize, pageSize);
    }

    @PreDestroy
    public void shutdown() {
        shutdownExecutor();
        log.info("数据清洗服务已关闭");
    }

    private void shutdownExecutor() {
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdownNow();
            try {
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    log.warn("线程池未能正常关闭");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            executorService = null;
        }
    }

    private void initExecutor() {
        // 创建新的线程池
        this.dataQueue = new LinkedBlockingQueue<>(maxQueueSize);
        this.executorService = new ThreadPoolExecutor(
                threadCount,
                threadCount * 2,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new ThreadFactory() {
                    private final AtomicInteger threadNum = new AtomicInteger(1);
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "data-clean-thread-" + threadNum.getAndIncrement());
                        thread.setDaemon(true);
                        return thread;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    @Override
    @DS("slave")
    @Deprecated
    public List<Map<String, Object>> fetchDataFromSource(String tableName) {
        log.warn("fetchDataFromSource 已废弃，建议使用分页方法 fetchDataByPage");
        return baseMapper.selectMaps(null);
    }

    @DS("slave")
    @Override
    public List<Map<String, Object>> fetchDataByPage(String tableName, String lastCtid, long pageSize) {
        if (!TABLE_NAME_PATTERN.matcher(tableName).matches()) {
            throw new IllegalArgumentException("非法表名: " + tableName);
        }

        QueryWrapper<Kdp01> queryWrapper = new QueryWrapper<>();
        if (lastCtid != null) {
            queryWrapper.apply("ctid > CAST({0} AS tid)", lastCtid);
        }
        queryWrapper
                .apply("EXISTS (SELECT 1 FROM kdp01 t2 WHERE t2.axcp0005 = kdp01.axcp0005 LIMIT 1)")
                .eq("axcp0077", "0")
                .eq("axcp0004", "01")
                .eq("is_delete", "0")
                .apply("LENGTH(axcp0005) = 18 AND axcp0005 NOT LIKE '% %'")
                .last("ORDER BY ctid LIMIT " + pageSize);

        return kdp01Mapper.selectMapsByWrapper(tableName, queryWrapper);
    }

    @Override
    public List<Map<String, Object>> cleanData(List<Map<String, Object>> rawData) {
        List<Map<String, Object>> cleanedData = new ArrayList<>(rawData.size());
        for (Map<String, Object> row : rawData) {
            Map<String, Object> cleanedRow = cleanSingleRow(row);
            if (cleanedRow != null) {
                cleanedData.add(cleanedRow);
            }
        }
        return cleanedData;
    }

    private Map<String, Object> cleanSingleRow(Map<String, Object> row) {
        // 移除空值
        row.values().removeIf(Objects::isNull);

        // 处理身份证号和年龄
        String idCard = (String) row.get("axcp0005");

        // 校验身份证号
        if (idCard == null || idCard.contains(" ")) {
            row.put("ageRange", "无效证件号（含空格）");
            return row;
        }

        if (idCard.length() != 18) {
            row.put("ageRange", "无效证件号（长度不符）");
            return row;
        }

        // 处理生日和年龄
        try {
            String birthStr = idCard.substring(6, 14);
            LocalDate birthDate = LocalDate.parse(birthStr, DateTimeFormatter.ofPattern("yyyyMMdd"));
            LocalDate now = LocalDate.now();
            int age = now.getYear() - birthDate.getYear();
            if (now.isBefore(birthDate.withYear(now.getYear()))) {
                age--;
            }

            row.put("birthDate", birthStr);
            row.put("age", age);
            row.put("ageRange", getAgeRange(age));
        } catch (Exception e) {
            row.put("ageRange", "无效证件号（格式错误）");
            log.warn("处理身份证号失败: {}", idCard, e);
        }

        return row;
    }

    @Override
    @DS("master")
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public boolean saveCleanedData(List<VolunteerInfo> cleanedData, String targetTable) {
        Set<String> batchProcessed = new HashSet<>();

        try {
            int size = cleanedData.size();
            for (int i = 0; i < size; i += batchSize) {
                int end = Math.min(i + batchSize, size);
                List<VolunteerInfo> batch = cleanedData.subList(i, end);

                // 过滤重复数据
                List<VolunteerInfo> filteredBatch = batch.stream()
                        .filter(info -> {
                            String idNumber = info.getIdNumber();
                            if (idNumber == null || batchProcessed.contains(idNumber)) {
                                return false;
                            }
                            batchProcessed.add(idNumber);
                            return true;
                        })
                        .collect(Collectors.toList());

                if (!filteredBatch.isEmpty()) {
                    batchCopyHelper.batchCopyInsert(targetTable, VolunteerInfo.class, filteredBatch);
                }
            }
            return true;
        } catch (Exception e) {
            log.error("保存清洗数据失败，目标表: {}", targetTable, e);
            throw new RuntimeException("保存清洗数据失败", e);
        }
    }

    @Override
    @DS("master")
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public boolean saveCleanedDataYoung(List<YoungsterInfo> youngsterInfoList, String targetTable) {
        try {
            int size = youngsterInfoList.size();
            for (int i = 0; i < size; i += batchSize) {
                int end = Math.min(i + batchSize, size);
                List<YoungsterInfo> batch = youngsterInfoList.subList(i, end);
                batchCopyHelper.batchCopyInsert(targetTable, YoungsterInfo.class, batch);
            }
            return true;
        } catch (Exception e) {
            log.error("插入青少年信息表失败，目标表: {}", targetTable, e);
            throw new RuntimeException("保存青少年数据失败", e);
        }
    }

    @Override
    @DS("master")
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public boolean saveCleanedDataTrajectory(List<VolunteerTrajectory> trajectoryInfoList, String targetTable) {
        try {
            int size = trajectoryInfoList.size();
            for (int i = 0; i < size; i += batchSize) {
                int end = Math.min(i + batchSize, size);
                List<VolunteerTrajectory> batch = trajectoryInfoList.subList(i, end);
                batchCopyHelper.batchCopyInsert(targetTable, VolunteerTrajectory.class, batch);
            }
            return true;
        } catch (Exception e) {
            log.error("插入轨迹信息表失败，目标表: {}", targetTable, e);
            throw new RuntimeException("保存轨迹数据失败", e);
        }
    }

    @Override
    public void executeDataCleanProcess(String sourceTable, String targetTable) throws InterruptedException {
        // 重置状态
        processedIdNumbers.get().clear();
        totalProcessedCount.set(0);
        totalBatchCount.set(0);

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        log.info("开始执行数据清洗流程: {} -> {}", sourceTable, targetTable);

        // 获取总记录数
        long totalCount = getTotalCount(sourceTable);
        long expectedBatches = (totalCount + pageSize - 1) / pageSize;
        log.info("源表总记录数: {}, 预计批次: {}", totalCount, expectedBatches);

        // 初始化线程池和计数器
        shutdownExecutor(); // 确保关闭之前的线程池
        initExecutor();
        processLatch = new CountDownLatch((int) expectedBatches);

        // 启动生产者线程
        startProducer(sourceTable);

        // 启动消费者线程
        startConsumers(targetTable);

        // 等待所有任务完成
        processLatch.await();

        // 等待线程池关闭
        shutdownExecutor();

        stopWatch.stop();
        log.info("数据清洗流程完成! 总耗时: {} 秒，处理批次: {}, 总记录数: {}",
                stopWatch.getTotalTimeSeconds(), totalBatchCount.get(), totalProcessedCount.get());
    }

    private void startProducer(String sourceTable) {
        executorService.submit(() -> {
            try {
                String lastCtid = null;
                while (true) {
                    List<Map<String, Object>> rawData = fetchDataByPage(sourceTable, lastCtid, pageSize);
                    if (rawData == null || rawData.isEmpty()) {
                        break;
                    }

                    // 放入队列
                    try {
                        dataQueue.put(rawData);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.warn("生产者线程被中断", e);
                        break;
                    }

                    // 更新最后一个CTID
                    try {
                        Map<String, Object> lastRecord = rawData.get(rawData.size() - 1);
                        Object ctidObj = lastRecord.get("ctid");
                        if (ctidObj != null) {
                            lastCtid = ctidObj.toString();
                        }
                    } catch (Exception e) {
                        log.error("获取CTID失败", e);
                    }

                    log.debug("已读取批次数据，记录数: {}", rawData.size());
                }
            } catch (Exception e) {
                log.error("生产者线程异常", e);
            } finally {
                // 发送结束信号
                for (int i = 0; i < threadCount; i++) {
                    try {
                        dataQueue.put(Collections.emptyList());
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
                log.info("生产者线程已结束");
            }
        });
    }

    private void startConsumers(String targetTable) {
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    while (true) {
                        List<Map<String, Object>> rawData = dataQueue.take();

                        // 检查结束标志
                        if (rawData.isEmpty()) {
                            break;
                        }

                        // 处理数据
                        try {
                            processBatchData(rawData, targetTable);
                        } catch (Exception e) {
                            log.error("批次处理失败", e);
                        } finally {
                            processLatch.countDown();
                            totalBatchCount.incrementAndGet();
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("消费者线程被中断", e);
                }
            });
        }
    }

    private void processBatchData(List<Map<String, Object>> rawData, String targetTable) {
        List<Map<String, Object>> cleanedData = cleanData(rawData);
        VolunteerInfoDTO volunteerInfoDTO = convertToVolunteerInfoList(cleanedData);

        // 保存数据
        boolean saved = saveCleanedData(volunteerInfoDTO.getVolunteerInfoList(), targetTable);

        // 保存青少年信息
        if (saved && !volunteerInfoDTO.getYoungsterInfoList().isEmpty()) {
            saveCleanedDataYoung(volunteerInfoDTO.getYoungsterInfoList(), "youngster_info");
        }

        // 保存轨迹信息
        if (saved && !volunteerInfoDTO.getVolunteerTrajectoryList().isEmpty()) {
            saveCleanedDataTrajectory(volunteerInfoDTO.getVolunteerTrajectoryList(), "volunteer_trajectory");
        }

        totalProcessedCount.addAndGet(cleanedData.size());
    }

    @DS("slave")
    private long getTotalCount(String tableName) {
        if (!TABLE_NAME_PATTERN.matcher(tableName).matches()) {
            throw new IllegalArgumentException("非法表名: " + tableName);
        }
        List<Map<String, Object>> resultMaps = kdp01Mapper.selectCounts(tableName);
        Map<String, Object> result = resultMaps.get(0);
        return Long.parseLong(result.get("count").toString());
    }

    private String getAgeRange(int age) {
        if (age >= 10 && age <= 23) {
            return "1";
        } else if (age < 60) {
            return "0";
        } else {
            return "2";
        }
    }

    public VolunteerInfoDTO convertToVolunteerInfoList(List<Map<String, Object>> cleanedData) {
        VolunteerInfoDTO volunteerInfoDTO = new VolunteerInfoDTO();
        List<VolunteerInfo> volunteerInfoList = new ArrayList<>();
        List<YoungsterInfo> youngsterInfoList = new ArrayList<>();
        List<VolunteerTrajectory> volunteerTrajectoryList = new ArrayList<>();

        Set<String> batchIdNumbers = new HashSet<>();

        for (Map<String, Object> row : cleanedData) {
            String idCard = (String) row.get("axcp0005");

            // 全局和当前批次双重检查
            if (idCard == null ||
                    processedIdNumbers.get().contains(idCard) ||
                    batchIdNumbers.contains(idCard)) {
                log.debug("跳过重复的身份证号: {}", idCard);
                continue;
            }
            processedIdNumbers.get().add(idCard);
            batchIdNumbers.add(idCard);

            VolunteerInfo volunteerInfo = new VolunteerInfo();
            YoungsterInfo youngsterInfo = new YoungsterInfo();

            // 安全转换各字段
            long seqValue = SnowflakeIdGeneratorUtils.getInstance().nextId();
            volunteerInfo.setId(seqValue);
            volunteerInfo.setName((String) row.get("axcp0002"));
            volunteerInfo.setIdType(convertToInteger(row.get("axcp0004")));
            volunteerInfo.setIdNumber(idCard);
            volunteerInfo.setIdImg(null);
            volunteerInfo.setSex(convertToInteger(row.get("axcp0006")));
            volunteerInfo.setNationality((String) row.get("axcp0007"));

            if (idCard != null && idCard.length() >= 14) {
                volunteerInfo.setDateOfBirth(idCard.substring(6, 14));
            } else {
                volunteerInfo.setDateOfBirth(" ");
            }

            volunteerInfo.setMobile((String) row.get("axcp0037"));
            volunteerInfo.setIsPublicVolunteerExperience(false);
            volunteerInfo.setServiceArea(String.valueOf(row.get("axaa0003")));
            volunteerInfo.setServiceMode(null);
            volunteerInfo.setTotalServiceHour(getTime((String) row.get("albp0289")));
            volunteerInfo.setTotalTrainingHour(Long.valueOf("0"));
            volunteerInfo.setServiceCategory((String) row.get("albp0206"));
            volunteerInfo.setServiceDomain(null);
            volunteerInfo.setCreateTime(new Date());
            volunteerInfo.setUpdateTime(new Date());
            volunteerInfo.setUsername(idCard);
            volunteerInfo.setPassword("1qaz@WSX");
            volunteerInfo.setCreateBy(null);
            volunteerInfo.setUpdateBy(null);
            volunteerInfo.setSalt("0");
            volunteerInfo.setStatus(convertToInteger(row.get("axcp0063")));
            volunteerInfo.setAvatar(null);
            volunteerInfo.setEmail((String) row.get("axzx0005"));
            volunteerInfo.setNickname(null);
            volunteerInfo.setCreatorName(null);
            volunteerInfo.setUpdatorName(null);
            volunteerInfo.setTelephone(null);
            volunteerInfo.setUnifiedCode(null);
            volunteerInfo.setReferral(null);
            volunteerInfo.setReferralUnifiedCode(null);
            volunteerInfo.setVolunteerStarLevel((short) 0);
            volunteerInfo.setScore(Float.valueOf("0.0"));
            volunteerInfo.setTabooStatus(convertToInteger(row.get("axcp0090")));
            volunteerInfo.setNation((String) row.get("axcp0008"));
            volunteerInfo.setAddress((String) row.get("axcp0090"));
            volunteerInfo.setIssuingOrganization(null);
            volunteerInfo.setStartDate(null);
            volunteerInfo.setExpiryDate(null);
            volunteerInfo.setPoliticsStatus((String) row.get("axcp0017"));
            volunteerInfo.setFirstLevelServiceField(null);
            volunteerInfo.setSecondLevelServiceField(null);
            volunteerInfo.setShowServiceHour(null);
            volunteerInfo.setAreaPath(null);
            volunteerInfo.setWxMiniOpenid(null);
            volunteerInfo.setUnionId(null);
            volunteerInfo.setWxOfficialOpenid(null);
            volunteerInfo.setValidPeriod(null);
            volunteerInfo.setIsIdChanged(null);
            volunteerInfo.setPlacePreference(null);
            volunteerInfo.setPlacePath(null);

            String age = getAgeRange(convertToInteger(row.get("age")));
            volunteerInfo.setVolunteerType(Integer.valueOf(age));

            if ("1".equals(age)) {
                long seqValueYoungster = SnowflakeIdGeneratorUtils.getInstance().nextId();
                youngsterInfo.setId(seqValueYoungster);
                youngsterInfo.setVolunteerInfoId(volunteerInfo.getId());
                String educationLevel = (String) row.get("axcp0009");
                if (educationLevel != null && !educationLevel.isEmpty()) {
                    youngsterInfo.setEducationLevel(Integer.valueOf(educationLevel));
                }
                youngsterInfoList.add(youngsterInfo);
            }

            volunteerInfo.setDeletionStatus(0);
            volunteerInfoList.add(volunteerInfo);

            VolunteerTrajectory volunteerTrajectory = new VolunteerTrajectory();
            long seqValueTrajectory = SnowflakeIdGeneratorUtils.getInstance().nextId();
            volunteerTrajectory.setId(seqValueTrajectory);
            volunteerTrajectory.setCode("REGISTER");
            volunteerTrajectory.setUserId(volunteerInfo.getId());
            volunteerTrajectory.setDescription("");
            volunteerTrajectory.setDetail("");
            volunteerTrajectory.setCreateTime(LocalDateTime.now());
            volunteerTrajectoryList.add(volunteerTrajectory);
        }

        volunteerInfoDTO.setVolunteerInfoList(volunteerInfoList);
        volunteerInfoDTO.setYoungsterInfoList(youngsterInfoList);
        volunteerInfoDTO.setVolunteerTrajectoryList(volunteerTrajectoryList);
        return volunteerInfoDTO;
    }

    private Long getTime(String time) {
        if (time == null || time.isEmpty()) {
            return 0L;
        }
        try {
            float f = Float.parseFloat(time);
            return (long) (f * 60);
        } catch (NumberFormatException e) {
            log.error("转换时间失败: {}", time, e);
            return 0L;
        }
    }

    private Integer convertToInteger(Object value) {
        if (value == null) {
            return null;
        }
        if (value instanceof Integer) {
            return (Integer) value;
        }
        try {
            return Integer.valueOf(value.toString());
        } catch (NumberFormatException e) {
            log.error("转换整数失败: {}", value, e);
            return null;
        }
    }
}